package iterator;

import global.AttrType;
import global.PageId;
import global.TupleOrder;
import heap.Heapfile;
import heap.Tuple;
import heap.exceptions.FieldNumberOutOfBoundException;
import heap.exceptions.InvalidTupleSizeException;
import heap.exceptions.InvalidTypeException;
import index.exceptions.IndexException;
import iterator.exceptions.JoinsException;
import iterator.exceptions.LowMemException;
import iterator.exceptions.PredEvalException;
import iterator.exceptions.SortException;
import iterator.exceptions.TupleUtilsException;
import iterator.exceptions.UnknowAttrType;
import iterator.exceptions.UnknownKeyTypeException;

import java.io.IOException;

import bufmgr.exceptions.PageNotReadException;

/**
 * The Sort class sorts a file. All necessary information are passed as
 * arguments to the constructor. After the constructor call, the user can
 * repeatly call <code>get_next()</code> to get tuples in sorted order. After
 * the sorting is done, the user should call <code>close()</code> to clean up.
 */
public class ExternalSort extends Sort {
	private static final int ARBIT_RUNS = 10;

	/**
	 * @uml.property name="_in"
	 * @uml.associationEnd multiplicity="(0 -1)"
	 */
	private AttrType[] _in;
	private short[] str_lens;
	private Iterator _am;
	private int _sort_fld;
	private TupleOrder order;
	private int _n_pages;
	private byte[][] bufs;
	private boolean first_time;
	private int Nruns;
	private int max_elems_in_heap;
	private int sortFldLen;
	private int tuple_size;

	private pnodeSplayPQ Q;
	/**
	 * @uml.property name="temp_files"
	 * @uml.associationEnd multiplicity="(0 -1)"
	 */
	private Heapfile[] temp_files;
	// private int n_tempfiles;
	// private Tuple output_tuple;
	private int[] n_tuples;
	private int n_runs;
	// private Tuple op_buf;
	private OBuf o_buf;
	/**
	 * @uml.property name="i_buf"
	 * @uml.associationEnd multiplicity="(0 -1)"
	 */
	private SpoofIbuf[] i_buf;
	/**
	 * @uml.property name="bufs_pids"
	 * @uml.associationEnd multiplicity="(0 -1)"
	 */
	private PageId[] bufs_pids;
	private boolean useBM = true; // flag for whether to use buffer manager

	/**
	 * Set up for merging the runs. Open an input buffer for each run, and
	 * insert the first element (min) from each run into a heap.
	 * <code>delete_min() </code> will then get the minimum of all runs.
	 * 
	 * @param tuple_size
	 *            size (in bytes) of each tuple
	 * @param n_R_runs
	 *            number of runs
	 * @exception IOException
	 *                from lower layers
	 * @exception LowMemException
	 *                there is not enough memory to sort in two passes (a
	 *                subclass of SortException).
	 * @exception SortException
	 *                something went wrong in the lower layer.
	 * @exception Exception
	 *                other exceptions
	 */
	private void merge() throws IOException, Exception {
		if (Nruns > _n_pages)
			throw new LowMemException(
					"Sort.java: Not enough memory to sort in two passes.");

		int i = 0;
		pnode cur_node;
		i_buf = new SpoofIbuf[Nruns];
		for (int j = 0; j < Nruns; j++)
			i_buf[j] = new SpoofIbuf();
		for (i = 0; i < Nruns; i++) {
			byte[][] apage = new byte[1][];
			apage[0] = bufs[i];

			// need iobufs.java
			i_buf[i].init(temp_files[i], apage, 1, tuple_size, n_tuples[i]);

			cur_node = new pnode();
			cur_node.run_num = i;

			// may need change depending on whether Get() returns the original
			// or make a copy of the tuple, need io_bufs.java ???
			Tuple temp_tuple = new Tuple(tuple_size);

			try {
				temp_tuple.setHdr(_in, str_lens);
			} catch (Exception e) {
				throw new SortException(e, "Sort.java: Tuple.setHdr() failed");
			}

			temp_tuple = i_buf[i].Get(temp_tuple); // need io_bufs.java

			if (temp_tuple != null) {
				/*
				 * System.out.print("Get tuple from run " + i);
				 * temp_tuple.print(_in);
				 */
				cur_node.tuple = temp_tuple; // no copy needed
				Q.enq(cur_node);

			}
		}
		return;
	}

	/**
	 * Generate sorted runs. Using heap sort.
	 * 
	 * @param max_elems
	 *            maximum number of elements in heap
	 * @param sortFldType
	 *            attribute type of the sort field
	 * @param sortFldLen
	 *            length of the sort field
	 * @return number of runs generated
	 * @exception IOException
	 *                from lower layers
	 * @exception SortException
	 *                something went wrong in the lower layer.
	 * @exception JoinsException
	 *                from <code>Iterator.get_next()</code>
	 */
	private void generate_runs(AttrType type) throws JoinsException,
			IndexException, InvalidTupleSizeException, InvalidTypeException,
			PageNotReadException, TupleUtilsException, PredEvalException,
			SortException, LowMemException, UnknowAttrType,
			UnknownKeyTypeException, IOException, Exception {

		pnodeSplayPQ currentQueue = new pnodeSplayPQ(_sort_fld, type, order);
		pnodeSplayPQ otherQueue = new pnodeSplayPQ(_sort_fld, type, order);

		Tuple LastTuple = new Tuple(tuple_size);
		Tuple dummyTuple;
		pnode currentNode;
		//
		try {
			LastTuple.setHdr(_in, str_lens);
		} catch (Exception e) {
			throw new SortException(e, "hena 3'alat");
		}
		int runNum = 0;
		// get min tuple or max
		if (order.tupleOrder == TupleOrder.Ascending) {
			getMin(LastTuple, type);
		} else {
			getMax(LastTuple, type);
		}

		int check;
		// fill current Queue
		while (currentQueue.size() + otherQueue.size() < max_elems_in_heap) {
			dummyTuple = _am.get_next();
			if (dummyTuple == null) {
				break;
			}
			pnode node = new pnode();
			node.tuple = new Tuple(dummyTuple);
			currentQueue.enq(node);
		}
		// write current queue

		while (true) {
			currentNode = currentQueue.deq();
			if (currentNode == null)
				break;

			check = TupleUtils.CompareTupleWithValue(type, currentNode.tuple,
					_sort_fld, LastTuple);

			if (order.tupleOrder == TupleOrder.Ascending && check < 0
					|| order.tupleOrder == TupleOrder.Descending && check > 0) {

				otherQueue.enq(currentNode);
			} else {
				// TupleUtils.SetValue(LastTuple, currentNode.tuple, _sort_fld,
				// type);
				LastTuple = currentNode.tuple;
				o_buf.Put(currentNode.tuple);
			}

			if (otherQueue.size() == max_elems_in_heap) {
				n_tuples[runNum] = (int) o_buf.flush();
				runNum++;

				if (runNum == n_runs) {
					Heapfile[] files = new Heapfile[n_runs * 2];
					for (int i = 0; i < n_runs; i++) {
						files[i] = temp_files[i];
					}
					temp_files = files;
					int[] tuples = new int[n_runs * 2];
					for (int i = 0; i < n_runs; i++) {
						tuples[i] = n_tuples[i];
					}
					n_tuples = tuples;

					n_runs *= 2;

				}
				Heapfile newFile;
				try {
					newFile = new Heapfile(null);
				} catch (Exception e) {
					throw new SortException(e,
							"Sort.java: create Heapfile failed");
				}
				temp_files[runNum] = newFile;

				// need io_bufs.java
				o_buf.init(bufs, _n_pages, tuple_size, temp_files[runNum],
						false);

				if (order.tupleOrder == TupleOrder.Ascending) {
					getMin(LastTuple, type);
				} else {
					getMax(LastTuple, type);
				}

				// switch the current heap and the other heap
				pnodeSplayPQ tempQ = currentQueue;
				currentQueue = otherQueue;
				otherQueue = tempQ;
			} else if (currentQueue.size() == 0) {
				while (currentQueue.size() + otherQueue.size() < max_elems_in_heap) {

					dummyTuple = _am.get_next();
					if (dummyTuple == null) {
						break;
					}
					pnode node = new pnode();
					node.tuple = new Tuple(dummyTuple);
					currentQueue.enq(node);

				}
			}
			if (currentQueue.size() == 0) {
				if (otherQueue.size() == 0)
					break;
				else {
					n_tuples[runNum] = (int) o_buf.flush();
					runNum++;
					if (runNum == n_runs) {
						Heapfile[] files = new Heapfile[n_runs * 2];
						for (int i = 0; i < n_runs; i++) {
							files[i] = temp_files[i];
						}
						temp_files = files;
						int[] tuples = new int[n_runs * 2];
						for (int i = 0; i < n_runs; i++) {
							tuples[i] = n_tuples[i];
						}
						n_tuples = tuples;

						n_runs *= 2;

					}
					Heapfile newFile;
					try {
						newFile = new Heapfile(null);
					} catch (Exception e) {
						throw new SortException(e,
								"Sort.java: create Heapfile failed");
					}
					temp_files[runNum] = newFile;

					// need io_bufs.java
					o_buf.init(bufs, _n_pages, tuple_size, temp_files[runNum],
							false);

					if (order.tupleOrder == TupleOrder.Ascending) {
						getMin(LastTuple, type);
					} else {
						getMax(LastTuple, type);
					}

					// switch the current heap and the other heap
					pnodeSplayPQ tempQ = currentQueue;
					currentQueue = otherQueue;
					otherQueue = tempQ;
				}
			}
		}
		n_tuples[runNum] = (int) o_buf.flush();
		runNum++;
		Nruns = runNum;
	}

	/**
	 * Remove the minimum value among all the runs.
	 * 
	 * @return the minimum tuple removed
	 * @exception IOException
	 *                from lower layers
	 * @exception SortException
	 *                something went wrong in the lower layer.
	 */
	private Tuple delete_min() throws IOException, Exception {
		pnode node = null;
		if (!Q.empty()) {
			node = Q.deq();
			pnode newNode = new pnode();
			newNode.run_num = node.run_num;
			Tuple new_tuple;
			if (i_buf[newNode.run_num].empty() != true) {
				new_tuple = new Tuple(tuple_size); // need tuple.java??

				try {
					new_tuple.setHdr(_in, str_lens);
				} catch (Exception e) {
					throw new SortException(e, "Sort.java: setHdr() failed");
				}
				new_tuple = i_buf[newNode.run_num].Get(new_tuple);
				if (new_tuple != null) {
					newNode.tuple = new_tuple;
					Q.enq(newNode);

				}

			}

		}
		if (node == null)
			return null;

		return node.tuple;
	}

	/**
	 * Set lastElem to be the minimum value of the appropriate type
	 * 
	 * @param lastElem
	 *            the tuple
	 * @param sortFldType
	 *            the sort field type
	 * @exception IOException
	 *                from lower layers
	 * @exception UnknowAttrType
	 *                attrSymbol or attrNull encountered
	 */
	private void getMin(Tuple lastElem, AttrType sortFldType)
			throws IOException, FieldNumberOutOfBoundException, UnknowAttrType {

		// short[] s_size = new short[Tuple.max_size]; // need Tuple.java
		// AttrType[] junk = new AttrType[1];
		// junk[0] = new AttrType(sortFldType.attrType);
		char[] c = new char[1];
		c[0] = Character.MIN_VALUE;
		String s = new String(c);
		// short fld_no = 1;

		switch (sortFldType.attrType) {
		case AttrType.attrInteger:
			// lastElem.setHdr(fld_no, junk, null);
			lastElem.setIntFld(_sort_fld, Integer.MIN_VALUE);
			break;
		case AttrType.attrReal:
			// lastElem.setHdr(fld-no, junk, null);
			lastElem.setFloFld(_sort_fld, Float.MIN_VALUE);
			break;
		case AttrType.attrString:
			// lastElem.setHdr(fld_no, junk, s_size);
			lastElem.setStrFld(_sort_fld, s);
			break;
		default:
			// don't know how to handle attrSymbol, attrNull
			// System.err.println("error in sort.java");
			throw new UnknowAttrType(
					"Sort.java: don't know how to handle attrSymbol, attrNull");
		}

		return;
	}

	/**
	 * Set lastElem to be the maximum value of the appropriate type
	 * 
	 * @param lastElem
	 *            the tuple
	 * @param sortFldType
	 *            the sort field type
	 * @exception IOException
	 *                from lower layers
	 * @exception UnknowAttrType
	 *                attrSymbol or attrNull encountered
	 */
	private void getMax(Tuple lastElem, AttrType sortFldType)
			throws IOException, FieldNumberOutOfBoundException, UnknowAttrType {

		// short[] s_size = new short[Tuple.max_size]; // need Tuple.java
		// AttrType[] junk = new AttrType[1];
		// junk[0] = new AttrType(sortFldType.attrType);
		char[] c = new char[1];
		c[0] = Character.MAX_VALUE;
		String s = new String(c);
		// short fld_no = 1;

		switch (sortFldType.attrType) {
		case AttrType.attrInteger:
			// lastElem.setHdr(fld_no, junk, null);
			lastElem.setIntFld(_sort_fld, Integer.MAX_VALUE);
			break;
		case AttrType.attrReal:
			// lastElem.setHdr(fld_no, junk, null);
			lastElem.setFloFld(_sort_fld, Float.MAX_VALUE);
			break;
		case AttrType.attrString:
			// lastElem.setHdr(fld_no, junk, s_size);
			lastElem.setStrFld(_sort_fld, s);
			break;
		default:
			// don't know how to handle attrSymbol, attrNull
			// System.err.println("error in sort.java");
			throw new UnknowAttrType(
					"Sort.java: don't know how to handle attrSymbol, attrNull");
		}

		return;
	}

	/**
	 * Class constructor, take information about the tuples, and set up the
	 * sorting
	 * 
	 * @param in
	 *            array containing attribute types of the relation
	 * @param len_in
	 *            number of columns in the relation
	 * @param str_sizes
	 *            array of sizes of string attributes
	 * @param am
	 *            an iterator for accessing the tuples
	 * @param sort_fld
	 *            the field number of the field to sort on
	 * @param sort_order
	 *            the sorting order (ASCENDING, DESCENDING)
	 * @param sort_field_len
	 *            the length of the sort field
	 * @param n_pages
	 *            amount of memory (in pages) available for sorting
	 * @exception IOException
	 *                from lower layers
	 * @exception SortException
	 *                something went wrong in the lower layer.
	 */
	public ExternalSort(AttrType[] in, short[] str_sizes, Iterator am,
			int sort_fld, TupleOrder sort_order, int sort_fld_len, int n_pages)
			throws IOException, SortException {
		super(in, str_sizes, am, sort_fld, sort_order, sort_fld_len, n_pages);
		_in = in;
		str_lens = str_sizes;
		Tuple t = new Tuple(); // need Tuple.java
		try {
			t.setHdr(_in, str_sizes);
		} catch (Exception e) {
			throw new SortException(e, "Sort.java: t.setHdr() failed");
		}
		tuple_size = t.size();
		_am = am;
		_sort_fld = sort_fld;
		order = sort_order;
		_n_pages = n_pages;
		first_time = true;
		max_elems_in_heap = 200;
		sortFldLen = sort_fld_len;
		// ------------------------------end of set values
		// generate ids of the buffer pole
		bufs_pids = new PageId[_n_pages];
		// each element in bufs is an page
		bufs = new byte[_n_pages][];

		// allocate pages in the buffer pool
		try {
			get_buffer_pages(_n_pages, bufs_pids, bufs);
		} catch (Exception e) {
			throw new SortException(e, "Sort.java: BUFmgr error");
		}
		// Initialize first heap file
		temp_files = new Heapfile[ARBIT_RUNS];
		n_tuples = new int[ARBIT_RUNS];
		n_runs = ARBIT_RUNS;

		try {
			temp_files[0] = new Heapfile(null);
		} catch (Exception e) {
			throw new SortException(e, "Sort.java: Heapfile error");
		}

		o_buf = new OBuf();
		// initialize o_buf
		o_buf.init(bufs, _n_pages, tuple_size, temp_files[0], false);

		Q = new pnodeSplayPQ(sort_fld, in[sort_fld - 1], order);
	}

	/**
	 * Returns the next tuple in sorted order. Note: You need to copy out the
	 * content of the tuple, otherwise it will be overwritten by the next
	 * <code>get_next()</code> call.
	 * 
	 * @return the next tuple, null if all tuples exhausted
	 * @exception IOException
	 *                from lower layers
	 * @exception SortException
	 *                something went wrong in the lower layer.
	 * @exception JoinsException
	 *                from <code>generate_runs()</code>.
	 * @exception UnknowAttrType
	 *                attribute type unknown
	 * @exception LowMemException
	 *                memory low exception
	 * @exception Exception
	 *                other exceptions
	 */
	public Tuple get_next() throws IOException, SortException, UnknowAttrType,
			LowMemException, JoinsException, Exception {
		if (first_time) {
			first_time = false;
			generate_runs(_in[_sort_fld - 1]);
			// setup_for_merge(tuple_size, Nruns);
			merge();
		}

		Tuple t = delete_min();
		if (t == null)
			return null;
		else
			return t;
	}

	/**
	 * Cleaning up, including releasing buffer pages from the buffer pool and
	 * removing temporary files from the database.
	 * 
	 * @exception IOException
	 *                from lower layers
	 * @exception SortException
	 *                something went wrong in the lower layer.
	 */
	public void close() throws SortException, IOException {
		// clean up
		if (!closeFlag) {

			try {
				_am.close();
			} catch (Exception e) {
				throw new SortException(e,
						"Sort.java: error in closing iterator.");
			}

			try {
				free_buffer_pages(_n_pages, bufs_pids);
			} catch (Exception e) {
				throw new SortException(e, "Sort.java: BUFmgr error");
			}
			for (int i = 0; i < _n_pages; i++)
				bufs_pids[i].pid = INVALID_PAGE;

			for (int i = 0; i < temp_files.length; i++) {
				if (temp_files[i] != null) {
					try {
						temp_files[i].deleteFile();
					} catch (Exception e) {
						throw new SortException(e, "Sort.java: Heapfile error");
					}
					temp_files[i] = null;
				}
			}
			closeFlag = true;
		}
	}

}
